package org.apm.data.pusher;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.util.StringUtil;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

public class DataPushTimer extends TimerTask {
    static final DateTimeFormatter TIME_FORMAT = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS");

    private static final Logger LOGGER = LogManager.getLogger(DataPushTimer.class);

    // private static final ScheduledExecutorService SINGLE =
    // Executors.newSingleThreadScheduledExecutor();
    private static final ScheduledExecutorService MULTI = Executors.newScheduledThreadPool(4);

    private static final int DAY = 24 * 60 * 60 * 1000;
    private static final int HOUR = 60 * 60 * 1000;
    private static final int FIVE_MINUTE = 5 * 60 * 1000;

    // private static final Client DEV_CLIENT = CommonMethod.devClient();
    private static final Client ONLINE_CLIENT = CommonMethod.onlineClient();
    private static final BulkProcessor DEV_PROCESSOR = CommonMethod.EsBulkProcessor.DEV.processor();
    // private static final BulkProcessor ONLINE_PROCESSOR =
    // CommonMethod.EsBulkProcessor.ONLINE.processor();

    private static final int SCROLL_TIME = 60000;
    private static final int SCROLL_SIZE = 1000;

    public DataPushTimer() {}

    /** the entry method. */
    public final void process() {
        LOGGER.info("run background.");
        Timer dailyTimer = new Timer();
        dailyTimer.scheduleAtFixedRate(this, 0, FIVE_MINUTE);
    }

    @Override
    public final void run() {

        String[] indices = { "apm2.0-115.238.29.251-httpdata-1", "apm2.0-115.238.29.251-mouseclick-1" };
        for (String index : indices) {
            MULTI.submit(new DataPushRunnable(index));
        }

    }

    private static String timestampField(String index) throws Exception {
        String type = index.split("-")[2];
        if (StringUtil.multEquals(type, "view", "snifferdata", "response", "relationship", "httpdata", "dbtype")) {
            return "timestamp";
        } else if (StringUtil.multEquals(type, "eventwaiting")) {
            return "createTime";
        } else if (StringUtil.multEquals(type, "javamanagementdata")) {
            return "jManagementMessage.creationTime";
        } else if (StringUtil.multEquals(type, "urlrecorddata")) {
            return "UrlRecordMessage.startTime";
        } else if (StringUtil.multEquals(type, "javatracedata")) {
            return "DefaultMessage.startTime";
        } else if (StringUtil.multEquals(type, "mouseclick")) {
            return "clickTime";
        } else if (StringUtil.multEquals(type, "pagemonitor")) {
            return "navigationStart";
        } else if (StringUtil.multEquals(type, "sessions")) {
            return "startTime";
        } else if (StringUtil.multEquals(type, "sniffersql", "resourcesmonitor", "event")) {
            return "";
        } else {
            throw new Exception("未知类型: " + index);
        }
    }

    public static void dataPush(final String masterIp, final String index, final String from, final String to) {
        String timestampField = "timestamp";
        try {
            timestampField = timestampField(index);
        } catch (Exception e) {
            return;
        }

        try {

            SearchRequestBuilder scrollRespBuilder = ONLINE_CLIENT.prepareSearch(index)
                    .setScroll(new TimeValue(SCROLL_TIME)).setSize(SCROLL_SIZE);
            if (StringUtil.isExist(timestampField)) {
                QueryBuilder query = QueryBuilders.boolQuery()
                        .filter(QueryBuilders.rangeQuery(timestampField).from(from).to(to));

                scrollRespBuilder.setQuery(query);
            }
            SearchResponse scrollResp = scrollRespBuilder.execute().actionGet();
            LOGGER.info("[Analyzer-After][" + index + "] " + scrollResp.getHits().getTotalHits() + ".");
            while (true) {
                if (scrollResp.getHits().getHits().length == 0) {
                    break;
                } else {
                    LOGGER.info("[Analyzer-After][" + index + "] " + scrollResp.getHits().getHits().length + ".");
                }

                for (SearchHit hit : scrollResp.getHits()) {
                    DEV_PROCESSOR.add(new IndexRequest(hit.getIndex(), hit.getType(), hit.getId())
                            .source(hit.getSourceAsString()));
                }

                scrollResp = ONLINE_CLIENT.prepareSearchScroll(scrollResp.getScrollId())
                        .setScroll(new TimeValue(SCROLL_TIME)).execute().actionGet();
            }
        } catch (Exception exception) {
            LOGGER.info(exception.getMessage());
        }
    }

    public static void main(String[] args) {
        new DataPushTimer().process();
    }

    private static class DataPushRunnable implements Runnable {

        String index;
        final String now = new DateTime().toString("yyyy-MM-dd HH:mm:ss");
        final String lastTime = new DateTime().minusMinutes(15).toString("yyyy-MM-dd HH:mm:ss");

        DataPushRunnable(final String index) {
            this.index = index;
        }

        @Override
        public void run() {
            String masterIp = index.split("-")[1];
            LOGGER.info("[Analyzer-After][" + index + "][" + masterIp + "] " + lastTime + " ---> " + now + ".");
            dataPush(masterIp, index, lastTime, now);
        }
    }
}
